#pragma once
#include "../common/net.hpp"
#include "../common/message.hpp"
#include<unordered_set>
namespace RPC_project
{
    namespace server
    {
        class TopicManager
        {
        public:
            using ptr = std::shared_ptr<TopicManager>;
            TopicManager(){}
            void onTopicRequest(const BaseConnection::ptr& conn ,const TopicRequest::ptr &msg)
            {
                //主题的创建
                //主题的删除
                //主题的订阅
                //主题的取消订阅
                //主题的发布
                TopicOptype topic_optype = msg->OpType();
                bool ret =true;
                switch(topic_optype)
                {
                    case TopicOptype::TOPIC_CREATE : topicCreate(conn ,msg);break;
                    case TopicOptype::TOPIC_REMOVE : topicRemove(conn ,msg);break;//取消主题
                    case TopicOptype::TOPIC_SUBSCRIBE :ret = topicSubscribe(conn ,msg);break;
                    case TopicOptype::TOPIC_CANCEL : topicCancel(conn ,msg);break;//取消订阅
                    case TopicOptype::TOPIC_PUBLISH : ret = topicPublish(conn ,msg);break;
                    default: return errorResponse(conn ,msg,RCode::RCODE_INVALID_OPTYPE);
                }
                if(!ret) return  errorResponse(conn ,msg,RCode::RCODE_NOT_FOUND_TOPIC);
                return topicResponse(conn ,msg);

            }
            //订阅者链接断开后，删除管理的数据
            void onShutdown(const BaseConnection::ptr &conn)
            {
                //消息发布者断开链接 ，没有操作
                //消息订阅者断开连接 ，需要删除其关联的数据
                //1. 判断断开连接的是否是订阅者，不是的话则直接返回
                //2. 获取因订阅者退出受影响的主题对象
                //3.从主题对象中移除该订阅者
                //4.从订阅者映射信息中删除订阅者
                std::vector<Topic::ptr> topics;
                Subscriber::ptr subscriber;
                {
                        std::unique_lock<std::mutex> lock(_mutex);
                        auto it = _subscribers.find(conn);
                        if(it == _subscribers.end())
                        {
                            //不是订阅者 
                            return;
                        }
                        subscriber = it->second;
                        for(auto &topic_name :subscriber->topics)
                        {
                            auto topic_it =_topics.find(topic_name);
                            if(topic_it == _topics.end()) continue;
                            topics.push_back(topic_it->second);
                        }
                        _subscribers.erase(it);
                }
                for(auto & topic :topics)
                {
                    topic->removeSubscriber(subscriber);
                }

            }
        private:
                void topicResponse(const BaseConnection::ptr &conn ,const TopicRequest::ptr &msg )
                {
                    auto base_rsp =MessageFactory::Create(MType::RSP_TOPIC);
                    RPC_project::TopicResponse::ptr msg_rsp= std::dynamic_pointer_cast<RPC_project::TopicResponse>(base_rsp);
                    msg_rsp->Setid(msg->Rid());
                    msg_rsp->SetRcode(RCode::RCODE_OK);
                    msg_rsp->SetMyType(MType::RSP_TOPIC);
                    conn->Send(msg_rsp);
                }

                void errorResponse(const BaseConnection::ptr &conn ,const TopicRequest::ptr &msg ,RCode rcode)
                {
                    auto base_rsp =MessageFactory::Create(MType::RSP_TOPIC);
                    RPC_project::TopicResponse::ptr msg_rsp= std::dynamic_pointer_cast<RPC_project::TopicResponse>(base_rsp);
                    msg_rsp->Setid(msg->Rid());
                    msg_rsp->SetRcode(rcode);
                    msg_rsp->SetMyType(MType::RSP_TOPIC);
                    conn->Send(msg_rsp);

                }

            void topicCreate(const BaseConnection::ptr& conn ,const TopicRequest::ptr &msg)
            {
                std::unique_lock<std::mutex> lock(_mutex);
                //构造一个主题对象，添加映射管理
                std::string topic_name =msg->TopickKey();
                auto topic = std::make_shared<Topic>(topic_name);
                _topics.insert(std::make_pair(topic_name ,topic));

            }

            void topicRemove(const BaseConnection::ptr& conn ,const TopicRequest::ptr &msg)
            {                //删除主题 ：1.找到当前主题的全部订阅者 ，从每个订阅者中将主题删除  2. 删除主题名称与主题对象的映射
                    std::string topic_name = msg->TopickKey();
                    std::unordered_set<Subscriber::ptr> subscribers;
                    {
                        std::unique_lock<std::mutex> lock(_mutex);
                        //在删除主题之前，先找出会受到影响的订阅者
                        auto it = _topics.find(topic_name);
                        if (it == _topics.end()) {
                            return;
                        }
                        subscribers= it->second->subscribers;

                        
                        _topics.erase(it);//删除当前的主题映射关系，
                    }
                    for (auto &subscriber : subscribers) {
                        subscriber->removeTopic(topic_name);
                    }  
            }

            //主题订阅
            bool topicSubscribe(const BaseConnection::ptr& conn ,const TopicRequest::ptr &msg)
            {
                //1.找出主题对象 ，以及订阅者对象
                //如果没有找到主题 ，就报错；如果没找到订阅者对象 ，就新构造一个订阅者
                Topic::ptr topic;
                Subscriber::ptr subscriber;
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    auto it_topic = _topics.find(msg->TopickKey());
                    if (it_topic == _topics.end()) {
                        return false;
                    }
                    topic= it_topic->second;

                    auto it_subscribe = _subscribers.find(conn);
                    if(it_subscribe != _subscribers.end())
                    {
                        subscriber =it_subscribe->second;
                    }
                    else
                    {
                        subscriber = std::make_shared<Subscriber>(conn);
                        _subscribers.insert(std::make_pair(conn ,subscriber));
                    }
                }

                //2.在主题对象中 ，新增一个订阅者对象关联的连接 ； 在订阅者对象中新增一个订阅的主题
                topic->appendSubscriber(subscriber);
                subscriber->appendTopic(msg->TopickKey());
                return true;
            }

            //取消订阅
            void topicCancel(const BaseConnection::ptr& conn ,const TopicRequest::ptr &msg)
            {
                //1.找出主体对象 ，和订阅者对象
                //主题不存在就报错 ，订阅者不存在不需要报错
                //2.在主题对象中 ，删除当前订阅者对象关联的连接 ； 在订阅者对象中删除此个订阅的主题
                Topic::ptr topic;
                Subscriber::ptr subscriber;
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    auto it_topic = _topics.find(msg->TopickKey());
                    if (it_topic != _topics.end()) {
                        topic= it_topic->second;
                    }

                    auto it_subscribe = _subscribers.find(conn);
                    if(it_subscribe != _subscribers.end())
                    {
                        subscriber = it_subscribe->second;
                    }
                }
                //2.在主题对象中 ，删除当前订阅者对象关联的连接 ； 在订阅者对象中删除此个订阅的主题
                if(subscriber) subscriber->removeTopic(msg->TopickKey());
                if(topic && subscriber) topic->removeSubscriber(subscriber);


            }

            bool topicPublish(const BaseConnection::ptr& conn ,const TopicRequest::ptr &msg)
            {
                Topic::ptr topic;
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    auto it_topic = _topics.find(msg->TopickKey());
                    if (it_topic == _topics.end()) {
                        return false;
                    }
                    topic= it_topic->second;
                }
                topic->pushMessage(msg);
                return true;
            }
        private:
            //订阅者
            struct Subscriber
            {
                using ptr = std::shared_ptr<Subscriber>;
                std::mutex _mutex;
                BaseConnection::ptr conn;
                std::unordered_set<std::string> topics;//订阅者订阅的多个主题
                Subscriber(const BaseConnection::ptr& connection):conn(connection)
                {}
                //订阅主题时调用
                void appendTopic(const std::string &topic_name)
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    topics.insert(topic_name);
                }
                //取消订阅或者主题被删除调用
                void removeTopic(const std::string &topic_name)
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    topics.erase(topic_name);
                }
            };

            struct Topic
            {
                using ptr = std::shared_ptr<Topic>;
                std::mutex _mutex;
                std::string topic_name;
                //查找效率是O(1) 
                std::unordered_set<Subscriber::ptr> subscribers;//主题的订阅者们的链接 ，主题收到消息利用set发送数据


                Topic(const std::string& topic_name): topic_name(topic_name){}

                //新增订阅者的时候调用
                void appendSubscriber(const Subscriber::ptr& subscriber)
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    subscribers.insert(subscriber);
                }
                //取消订阅 或者 订阅者连接断开时调用
                void removeSubscriber(const Subscriber::ptr& subscriber)
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    subscribers.erase(subscriber);
                }
                //收到消息发布请求 ，将消息序列化用链接set发送
                void pushMessage(const BaseMessage::ptr &msg)
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    for(auto subscriber: subscribers)
                    {
                        subscriber->conn->Send(msg);
                    }
                }
            };

           
        private:
            std::mutex _mutex;
            std::unordered_map<std::string ,Topic::ptr>      _topics;//主题名称与主题的映射关系
            std::unordered_map<BaseConnection::ptr ,Subscriber::ptr>      _subscribers;//链接与订阅者的映射 
        };
    }
}